package com.dhl.springbootrabbitmq.mqcallback;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

/**
 * 高级篇消息确认机制
 * 配置文件需要开启下面配置
 * spring.rabbitmq.publisher-confirm-type=correlated
 *
 * @author danghailiang
 */
@Slf4j
@Component
public class CustomCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 注入替换原有接口
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        /**
         * true：
         * 交换机无法将消息进行路由时，会将该消息返回给生产者
         * false：
         * 如果发现消息无法进行路由，则直接丢弃
         */
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * 交换机挂掉之后消息确认回调
     *
     * @param correlationData 发送的数据
     * @param ask             是否成功
     * @param errorMsg        成功时：null，不成功时：失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ask, String errorMsg) {
        // 获取消息id
        String messageId = ObjectUtils.isEmpty(correlationData) ? "无消息id" : correlationData.getId();
        if (ask) {
            log.info("消息发送成功，消息id为：{}", messageId);
        } else {
            log.error("消息发送失败：失败id：{}，失败原因：{}", messageId, errorMsg);
        }
    }

    /**
     * 队列消息发送回调
     *
     * @param returnedMessage
     */
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        log.warn("message:{}", new String(message.getBody(), StandardCharsets.UTF_8));
        log.warn("ReplyText:{}", returnedMessage.getReplyText());
        log.warn("ReplyCode:{}", returnedMessage.getReplyCode());
        log.warn("Exchange:{}", returnedMessage.getExchange());
        log.warn("RoutingKey:{}", returnedMessage.getRoutingKey());
    }
}
